package com.bw.gmall;

import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.alibaba.ververica.cdc.debezium.StringDebeziumDeserializationSchema;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class FlinkCDCSQL {
    public static void main(String[] args) throws Exception {
        //1.创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //        //开启CK，创建文件夹
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
//        // 每隔5秒保存一次
//        env.enableCheckpointing(5000L);
//        // 精准消费一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        // 配置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(100000L);
//        // 同时最大可以运行几个ck
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        // 上一个ck的尾部到下一个ck的头之间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        //2.创建动态表
         // 创建表的执行环境
        StreamTableEnvironment streamTableEnvironment = StreamTableEnvironment.create(env);
        streamTableEnvironment.executeSql("CREATE TABLE mysql_binlog (" +
                "  id STRING NOT NULL," +
                "  tm_name STRING," +
                "  logo_url STRING" +
                ") WITH (" +
                "  'connector' = 'mysql-cdc'," +
                "  'hostname' = 'hadoop102'," +
                "  'port' = '3306'," +
                "  'username' = 'root'," +
                "  'password' = '123456'," +
                "  'database-name' = 'gmall-flink'," +
                "  'table-name' = 'base_trademark'" +
                ")");
        //3.查询数据
        Table table = streamTableEnvironment.sqlQuery("select * from mysql_binlog");
        //4.把表转成流
        //toAppendStream与toRetractStream的区别 https://www.jianshu.com/p/ee7b4a1004e4
        DataStream<Tuple2<Boolean, Row>> retractstream = streamTableEnvironment.toRetractStream(table, Row.class);
        //5.打印数据
        retractstream.print();
        //6.执行
        env.execute("FlinkCDCSQL");


    }
}
